package cn.lingyangwl.agile.msg.push.listener;

import cn.lingyangwl.agile.msg.push.api.model.mq.MsgPushMQ;
import cn.lingyangwl.agile.msg.push.api.model.rqrs.MsgPushParam;
import cn.lingyangwl.agile.msg.push.core.propcess.MsgPushProcess;
import cn.lingyangwl.framework.mq.base.manager.MqManager;
import cn.lingyangwl.framework.tool.core.exception.BizException;
import com.alibaba.fastjson2.JSON;
import com.rabbitmq.client.Channel;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;

/**
 * 延时消息超时处理
 * 死信交换机
 * 定义死信队列
 * @author shenguangyang
 */
@Slf4j
@Component
@RabbitListener(queues = MsgPushMQ.DELAYED_QUEUE)
public class PushDelayedListener {
    @Resource
    private MqManager mqManager;
    @Resource
    private MsgPushProcess msgPushHandler;

    @RabbitHandler
    public void onMessage(String data, Channel channel, Message message) {
        try {
            MsgPushMQ.MsgPayload msgPayload = MsgPushMQ.parse(data, MsgPushMQ.MsgPayload.class);
            // 判断是否重复消费
            if (mqManager.isConsumed(msgPayload.getMsgId())) {
                log.warn("message {} are repeatedly consumed, do not perform business", msgPayload.getMsgId());
                return;
            }
            MsgPushParam msgPushParam = msgPayload.getMsgPushParam();
            msgPushHandler.push(msgPushParam, msgPayload.getMsgId());
            log.info("消费成功: {}", JSON.toJSONString(data));
            mqManager.markConsumed(msgPayload.getMsgId());
        } catch (Exception e) {
            log.error("消费失败: {}, msg: {}", e.getMessage(), JSON.toJSONString(data));
            if (e instanceof NullPointerException) {
                e.printStackTrace();
            }
            throw new BizException(-1, "mq消费错误");
        }
    }
}
